package com.zyy.study.jdk8NewFeatures;

import io.netty.util.concurrent.DefaultThreadFactory;
import jodd.util.concurrent.ThreadFactoryBuilder;
import org.junit.Test;

import java.util.concurrent.*;
import java.util.concurrent.CompletableFuture;


public class CompletableFutureDemo {
    private static final Executor executor = new ThreadPoolExecutor(8, 16, 1, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

    /**
     * 创建异步任务，线程池是我们自定义得
     */
    @Test
    public void testSupplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            return 1 + 1;
        }, executor);
        //推荐使用join获取异步结果
        System.out.println(integerCompletableFuture.join());
        System.out.println(integerCompletableFuture.get());
    }

    /**
     * thenApply接收上个任务的结果继续处理
     */
    @Test
    public void testApply() {
        Integer result = CompletableFuture.supplyAsync(() -> {
            return 1 + 2;
        }, executor).thenApply(r -> {
            return r + 3;
        }).join();
        System.out.println(result);
    }

    /**
     * result:
     * ForkJoinPool.commonPool-worker-1：执行任务1
     * ForkJoinPool.commonPool-worker-1：执行任务2
     * 2
     * CompletableFuture不传线程池默认使用ForkJoinPool.commonPool
     */
    @Test
    public void testApply2() {
        Integer join = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "：执行任务1");
            return 1;
        }).thenApply((r1) -> {
            System.out.println(Thread.currentThread().getName() + "：执行任务2");
            return r1 + 1;
        }).join();
        System.out.println(join);
    }

    /**
     * result:
     * pool-1-thread-1：执行任务1
     * pool-1-thread-1：执行任务2
     * 2
     * thenApply复用上个任务的线程
     */
    @Test
    public void testApply3() {
        Integer join = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "：执行任务1");
            return 1;
        }, executor).thenApply((r1) -> {
            System.out.println(Thread.currentThread().getName() + "：执行任务2");
            return r1 + 1;
        }).join();
        System.out.println(join);
    }

    /**
     * pool-1-thread-1：执行任务1
     * pool-1-thread-2：执行任务2
     * thenApplyAsync传线程池，怎么使用新的线程执行任务，否则使用ForkJoinPool.commonPool
     */
    @Test
    public void testApply4() {
        Integer join = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "：执行任务1");
            return 1;
        }, executor).thenApplyAsync((r1) -> {
            System.out.println(Thread.currentThread().getName() + "：执行任务2");
            return r1 + 1;
        }, executor).join();
    }

    /**
     * 通过thenCombine两两合并结果
     *method1异常java.lang.ArithmeticException: / by zero
     * pool-1-thread-1开始合并结果
     * pool-1-thread-1合并结果
     * result:5
     */
    @Test
    public void testCombine1() {
        CompletableExample completableExample = new CompletableExample();
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            return completableExample.method1();
        }, executor).exceptionally(e -> {
            System.out.println("method1异常" + e.getMessage());
            return null;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            return completableExample.method2();
        }, executor).exceptionally(e -> {
            System.out.println("method2异常" + e.getMessage());
            return null;
        });
        CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
            return completableExample.method3();
        }, executor).exceptionally(e -> {
            System.out.println("method3异常" + e.getMessage());
            return null;
        });

        try {
            Integer integer = future1.thenCombine(future2, (f1, f2) -> {
                System.out.println(Thread.currentThread().getName() + "开始合并结果");
                if (f1 == null||f2==null) {
                    return f1 == null ? f2 : f1;
                }
                return f1 + f2;
            }).thenCombine(future3, (r1, f3) -> {
                System.out.println(Thread.currentThread().getName() + "合并结果");
                return r1 + f3;
            }).get(2, TimeUnit.SECONDS);
            System.out.println("result:" + integer);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
            System.out.println("获取结果超时");
        }
    }

}

class CompletableExample {
    public int method1() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        int i = 1 / 0;
        return 1;
    }

    public int method2() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 2;
    }

    public static int method3() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 3;
    }
}
